package com.chenjj.java.mq.rabbitmq;

import cn.hutool.json.JSONUtil;
import com.chenjj.java.business.order.OrderInfoVO;
import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.URISyntaxException;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.Random;
import java.util.concurrent.TimeoutException;

/**
 * 路由消费者
 */
public class OrderConsumerMailRouting {
    private Logger logger = LoggerFactory.getLogger(OrderProviderSimple.class);

    private String defaultUri = "amqp://cjj:cjj@localhost:5672/vhost_cjj";

    private ConnectionFactory connectionFactory = new ConnectionFactory();

    public OrderConsumerMailRouting() throws NoSuchAlgorithmException, KeyManagementException, URISyntaxException {
        connectionFactory.setUri(defaultUri);
    }

    public OrderConsumerMailRouting(String uri) throws NoSuchAlgorithmException, KeyManagementException, URISyntaxException {
        connectionFactory.setUri(uri);
    }

    /**
     * 从队列中取一条订单信息并发送邮件
     */
    public void consumerOrderInfo() {
        String exchangeName = "orderInfoExchangeDirect";
        String routeKey_VIP = "vip"; //vip 消费金额大于5000的同时发邮件和短信提醒
        String routeKey_NORMAL = "normal"; //normal 消费金额小于5000的只发短信

        Connection connection = null;
        Channel channel = null;
        try {
            connection = connectionFactory.newConnection();
            channel = connection.createChannel();
            //声明一个临时队列，该队列在连接断开以后会自动删除
            String queueName = channel.queueDeclare().getQueue();
            //声明一个direct类型的、持久化的exchange
            channel.exchangeDeclare(exchangeName,"direct",true,false,null);
            //将队列和exchange绑定，只关注vip的数据
            channel.queueBind(queueName,exchangeName, routeKey_VIP);
            //一次取一条
            channel.basicQos(2);
            Consumer orderConsumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message = new String(body, "UTF-8");

                    OrderInfoVO orderInfoVO = JSONUtil.toBean(message,OrderInfoVO.class);

                    try{
                        if(new Random().nextInt(10)>=5){
                            throw new RuntimeException("业务异常");
                        }

                        System.out.println("Consumer Received '" + message + "' , 准备发送邮件");

                        //发送短信
                        //do ...

                        //向Rabbit Server发送ack，确认消息已经消费成功，server接受到ack之后会将该消息删除
                        getChannel().basicAck(envelope.getDeliveryTag(),false);
                    }catch (Exception e){
                        System.err.println("业务异常，退回消息，重新消费");
                        //通知server，此条消息没有处理成功，重新放回队列，需要再次被消费。
                        //如果不显示的reject，那么该条消息会被一直放在uacked队列，直到连接释放之后，才会重新放回队列，
                        //且处理完prefetchCount的消息之后，不会再继续消费新的消息。
                        //requeue:是否重新放入队列，true:是，false:否，且会丢弃该条消息
                        getChannel().basicReject(envelope.getDeliveryTag(),true);
                    }finally {}
                }
            };
            channel.basicConsume(queueName,false,orderConsumer);

            while(true){
                Thread.sleep(1);
            }
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("", e);
        } finally {
            releaseResource(connection, channel);
        }
    }

    private void releaseResource(Connection connection,Channel channel){
        if (channel!=null){
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
        }
        if(connection!=null){
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

}
